package bsp.diff;

import java.io.IOException;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hama.HamaConfiguration;
import org.apache.hama.bsp.BSPPeer;
import org.apache.hama.bsp.FileInputFormat;
import org.apache.hama.bsp.FileOutputFormat;
import org.apache.hama.bsp.TextInputFormat;
import org.apache.hama.bsp.TextOutputFormat;
import org.apache.hama.bsp.sync.SyncException;

import util.DiffComputation;
import util.DiffJobA;
import util.Parsable;
import bsp.BspHelper;
import bsp.CMapper;
import bsp.CReducer;

public abstract class BspDiffHelper<T0 extends Parsable<T0>, T1 extends Parsable<T1>, T2 extends Parsable<T2>>
		extends BspHelper<LongWritable, Text, Text, NullWritable, Text> {



	public static DiffJobA createScanJob(final Path in, final Path out)
			throws IOException {
		HamaConfiguration conf = new HamaConfiguration();
		DiffJobA job = new DiffJobA(conf);
		job.setInputFormat(TextInputFormat.class);
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(NullWritable.class);
		job.setOutputFormat(TextOutputFormat.class);
		FileInputFormat.setInputPaths(job, in);
		FileOutputFormat.setOutputPath(job, out);
		// jobClient = new BSPJobClient(job.getConfiguration());
		// cluster = jobClient.getClusterStatus(true);

		return job;
	}



	protected DiffComputation<T0, T1, T2> diff;

	public BspDiffHelper() {
		// BspDiffHelper(cons_c);
		setDiffComputation();
	}

	@Override
	public void bsp(
			BSPPeer<LongWritable, Text, Text, NullWritable, Text> peer)
			throws IOException, SyncException, InterruptedException {

		int count = 0;
		
		CMapper<LongWritable, Text, Text, NullWritable, Text> preMapper = null;
		for (CMapper<LongWritable, Text, Text, NullWritable, Text> mapper : mappers) {
			System.out.println("*****map*****" + count +" --" + peer.getPeerName());
			mapper.setup(peer);
			mapper.map(peer);
			// TODO: add to cache
			mapper.sendAllMesgs(peer);
			preMapper = mapper;
			// update cache and output
			// peer.clear(); // TODO just for test
			peer.reopenInput();
			count ++;
			peer.sync();
		}

		for (CReducer<LongWritable, Text, Text, NullWritable, Text> reducer : reducers) {
			System.out.println("*****reduce*****" + peer.getPeerName());
			reducer.setup(peer);
			reducer.reduce(peer);
			reducer.sendAllMesgs(peer);
			peer.sync();
		}

	}
	
	@Override
	public void setAllCMappers() {

		this.addMapper(new DiffStepOne<T0, T1, T2>(diff)); // oplus
		this.addMapper(new DiffStepTwo<T0, T1, T2>(diff));
		 
		this.addMapper(new DiffStepThree<T0, T1, T2>(diff));

	}
	
	@Override
	public void setAllCReducers() {
		// if(reducers == null)
		// reducers = new ArrayList<CReducer<LongWritable, Text, Text,
		// NullWritable, DiffMessage>>();

		//this.addReducer(new DiffStepTwo<T0, T1, T2>(diff));
	}
	

	public abstract void setDiffComputation(); // set DiffComputation<T0,T1,T2>
}